package me.tuyou.wordcount;

import me.tuyou.common.LineSplitter;
import me.tuyou.common.WordStreamGenerator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 单词统计(流处理)
 */
public class StreamWordCountTest {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setStateBackend(new MemoryStateBackend());
        DataStream<Tuple2<String, Integer>> dataStream = env.addSource(new WordStreamGenerator())
                .flatMap(new LineSplitter())
                .keyBy(item -> item.f0)
                .timeWindow(Time.seconds(3))
                .sum(1);

        dataStream.print("wordCount");

        //打印执行计划，可视化：https://flink.apache.org/visualizer/
        System.out.println(env.getExecutionPlan());

        env.execute("wordCount");
    }
}
